package com.example;


import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.opencsv.CSVReader;

// kafka生产者代码
public class KProducer {
  public static long count = 0;
  public static long rate = 0;

  public static void main(String[] args) throws InterruptedException, IOException {
    Properties props = new Properties();
    // kafka 集群，broker-list
    // props.put("bootstrap.servers", "121.43.165.220:9092");
    props.put("bootstrap.servers", "127.0.0.1:9092");
    props.put("acks", "0");
    // 重试次数
    props.put("retries", 1);
    // 批次大小
    props.put("batch.size", 16384);
    // 等待时间
    props.put("linger.ms", 0);
    // RecordAccumulator 缓冲区大小
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    Producer<String, String> producer = new KafkaProducer<>(props);
    CSVReader cReader = new CSVReader(new FileReader("D:\\Download\\records.csv"));
    Timer timer = new Timer();
    timer.scheduleAtFixedRate(new TimerTask() {
      public void run() {
        System.out.println("send " + rate  + " record/s, total is " + count);
        rate = 0;
      }
    }, 1000, 1000);
    // while(true){
    //   Thread.sleep(1000);
    //   ProducerRecord<String, String> record = new ProducerRecord<>("0323", String.valueOf(count));
    //   producer.send(record);
    //   count++;
    //   rate++;
    // }
    for(String[] c : cReader){
      try {
        // 生产单条数据示例
        // System.out.println(c[0]);
        // Thread.sleep(2000);
        String recordString = new String(c[0].getBytes(), "gb2312");
        ProducerRecord<String, String> record = new ProducerRecord<>("0401", recordString);
        producer.send(record);
        count++;
        rate++;
      }
      catch (Exception e){
        System.out.println(e);
      }
    }
  }
}